package kafka_consumer.util;

import java.io.IOException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import kafka_consumer.entity.ConsumerEntity;
import kafka_consumer.service.ConsumerService;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.stereotype.Component;
@Component
public class ConsumerUtil {
	@Autowired
	private ObjectMapper objectMapper;// ObjectMapper类是Jackson库的主要类。它提供一些功能将转换成Java对象匹配JSON结构，反之亦然。
	@Autowired
	private ConsumerService consumerService;

	// 在方法上使用@KafkaListener注解，并指定要消费的topic（也可以指定消费组以及分区号，支持正则表达式匹配），这样，消费者一旦启动，就会监听kafka服务器上的topic，实时进行消费消息
	@KafkaListener(topics = { "kafka-YL" })
	public void insert(String message) throws IOException {
//		读取生产者的所有数据
		ConsumerEntity[] ces = objectMapper.readValue(message, ConsumerEntity[].class);
		for(ConsumerEntity ce:ces) {
			System.out.println("获取姓名: " + ce.getName());
			System.out.println("已消费消息:" + message);
			consumerService.insertService(ce.getId(),ce.getName(),ce.getAge());
		}
	}
}
